package com.rem.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * 发送数据到kafka
 *
 * @author Rem
 * @date 2022-12-28
 */
public class CustomerProducer {

    public static void main(String[] args) {
        Properties properties = new Properties();
        //连接kafka
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9093,localhost:9094,localhost:9095");
        //序列化
//        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
//        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        //RecordAccumulator 缓冲区总大小，默认 32m
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 32 * 1024 * 1024);
        //缓冲区一批数据最大值，默认 16k
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16 * 1024);
        //如果数据迟迟未达到 batch.size，sender 等待 linger.time之后就会发送数据
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 0);
        //0：生产者发送过来的数据，不需要等数据落盘应答
        //1：生产者发送过来的数据，Leader 收到数据后应答
        //-1（all）：生产者发送过来的数据
        properties.put(ProducerConfig.ACKS_CONFIG, "-1");
        //允许最多没有返回 ack 的次数，默认为 5 开启幂等性要保证该值是 1-5 的数字
        properties.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
        //当消息发送出现错误的时候，系统会重发消息 重试次数。默认是 int 最大值，2147483647
        properties.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
        //两次重试之间的时间间隔，默认是 100ms
        properties.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 100);
        //是否开启幂等性，默认 true，开启幂等性
        properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        //生产者发送的所有数据的压缩方式。默认是 none，也就是不压缩 none、gzip、snappy、lz4 和 zstd。
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "none");

        //配置kafka连接器以及key value序列化方式
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties, new StringSerializer(), new StringSerializer());

        for (int i = 0; i < 5000; i++) {
            kafkaProducer.send(new ProducerRecord<>("first", "hello" + i));
        }
        kafkaProducer.close();
    }
}
